﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Lucandra.Net.Connection.Pooling;
using Lucandra.Net.Connection;
using Lucandra.Net.Connection.Endpoint;
using Lucandra.Configuration;
using Apache.Cassandra;
using System.Diagnostics;

namespace Lucandra.Net
{
    internal static class KeyspaceManager
    {
        private static HashSet<Keyspace> actualKeyspaces;
        private static Dictionary<string, string> configuredKeyspaces;
        private static bool isInitialized;

        #region Type Initializer

        static KeyspaceManager()
        {
            actualKeyspaces = new HashSet<Keyspace>();
            configuredKeyspaces = new Dictionary<string, string>();
        }

        #endregion

        #region Methods [private] [static]

        /// <summary>
        /// Initializes the keyspace manager.  Caches information from the configuration file and pulls/caches schemas from Cassandra.
        /// </summary>
        private static void Initialize()
        {
            if (isInitialized)
                return;

            var config = LucandraConfigurationSection.GetConfig();

            foreach (IndexElement index in config.Indexes)
                configuredKeyspaces[index.Name] = index.Keyspace;

            var endpoint = new CassandraEndpoint(config.LiveSchemaUpdateNode.Host, config.LiveSchemaUpdateNode.Port, config.LiveSchemaUpdateNode.Timeout);
            var client = CassandraConnectionFactory.Create(endpoint);
            client.OpenTransport();

            var cassandraKeyspaces = client.InnerClient.describe_keyspaces();
            foreach (var keyspace in cassandraKeyspaces)
            {
                var cfs = new List<ColumnFamily>();
                foreach(var cfdef in keyspace.Cf_defs)
                {
                    ColumnFamilyType cfType;
                    if (cfdef.Column_type == "Super")
                        cfType = ColumnFamilyType.Super;
                    else if (cfdef.Column_type == "Standard")
                        cfType = ColumnFamilyType.Standard;
                    else
                        throw new LucandraException("Unknown column family type: " + cfdef.Column_type);

                    cfs.Add(new ColumnFamily(cfdef.Name,cfType));
                }
                var ks = new Keyspace(keyspace.Name, cfs);

                actualKeyspaces.Add(ks);
            }

            isInitialized = true;
        }

        /// <summary>
        /// Asserts that the keyspace manager is initialized.  If the keyspace manager has not been initialized, an initialization is performed.
        /// </summary>
        private static void AssertIsInitialized()
        {
            if (!isInitialized)
                Initialize();
        }

        /// <summary>
        /// Gets the configuration element for the specified index.
        /// </summary>
        /// <param name="indexName">The name of the index.</param>
        /// <returns>The configuration element for the index.</returns>
        private static IndexElement GetIndexElement(string indexName)
        {
            IndexElement element = null;
            foreach (IndexElement e in LucandraConfigurationSection.GetConfig().Indexes)
            {
                if (e.Name == indexName)
                {
                    element = e;
                    break;
                }
            }

            if (element == null)
                throw new LucandraException("The index `" + indexName + "` does not exist in the Lucandra configuration.");

            return element;
        }

        #endregion

        public static void AssertKeyspaceExists(string indexName)
        {
            var keyspaceName = GetKeyspaceFromIndexName(indexName);

            var actualKeyspace = actualKeyspaces
                .Where(ks => ks.Name == keyspaceName)
                .Where(ks => ks.ColumnFamilies.Where(cf => cf.Name == Settings.DocumentsColumnFamily || cf.Name == Settings.TermInfoColumnFamily).Count() == 2)
                .SingleOrDefault();

            if (actualKeyspace == null)
                CreateKeyspaceForIndex(indexName);
        }

        public static string GetKeyspaceFromIndexName(string indexName)
        {
            AssertIsInitialized();

            if (Settings.UseKeyspacePerIndex)
            {
                string keyspace;
                if (!configuredKeyspaces.TryGetValue(indexName, out keyspace))
                    throw new LucandraException("There is no keyspace configured for the index `" + indexName + "`.");
                return keyspace;
            }
            else
                return Settings.Keyspace;
        }

        public static void CreateKeyspaceForIndex(string indexName)
        {
            AssertIsInitialized();

            var indexConfig = GetIndexElement(indexName);

            int replicationFactor = 2;
            string replicationStrategy = "org.apache.cassandra.locator.SimpleStrategy";

            if (indexConfig.AutoCreateSettings != null)
            {
                replicationFactor = indexConfig.AutoCreateSettings.ReplicationFactor;
                replicationStrategy = indexConfig.AutoCreateSettings.ReplicationStrategy;
            }

            try
            {
                using (var client = CassandraConnectionPool.Borrow())
                {
                    var keyspaces = client.InnerClient.describe_keyspaces().ToArray();
                    var keyspace = keyspaces.Where(ks => ks.Name == indexName).SingleOrDefault();

                    if (keyspace == null)
                    {
                        keyspace = new KsDef()
                        {
                            Name = indexConfig.Keyspace,
                            Replication_factor = replicationFactor,
                            Strategy_class = replicationStrategy,
                            Cf_defs = new List<CfDef>()
                        };

                        var docCF = new CfDef()
                        {
                            Name = Settings.DocumentsColumnFamily,
                            Column_type = "Super",
                            Comparator_type = "BytesType",
                            Subcomparator_type = "BytesType",
                            Row_cache_size = 10000,
                            Key_cache_size = 0.05,
                            Keyspace = indexConfig.Keyspace
                        };
                        //client.InnerClient.system_add_column_family(docCF);
                        keyspace.Cf_defs.Add(docCF);

                        var termsCF = new CfDef()
                        {
                            Name = Settings.TermInfoColumnFamily,
                            Column_type = "Standard",
                            Comparator_type = "BytesType",
                            Row_cache_size = 0.1,
                            Key_cache_size = 0.1,
                            Keyspace = indexConfig.Keyspace
                        };
                        //client.InnerClient.system_add_column_family(termsCF);
                        keyspace.Cf_defs.Add(termsCF);

                        client.InnerClient.system_add_keyspace(keyspace);
                    }
                }
            }
            catch (Apache.Cassandra.InvalidRequestException ex)
            {
                Debug.WriteLine(ex);
                Debug.WriteLine("Why: " + ex.Why);
                throw;
            }
            catch (Exception ex)
            {
                Debug.WriteLine(ex);
                throw;
            }
        }
    }
}
